package com.we.risk.attribution.exposure;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.risk.attribution.AttributionInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Optional;
import java.util.Properties;

import static com.we.risk.attribution.AttributionInfo.*;

public class AdmExposureAttributionKfkLabel {
    public static final String RELEASEPROP =
            "risk/attribution/exposure/kfk_exposure_attribution_test.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmExposureAttributionKfkLabel.class);

    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String LABELNAME = "label_name";
    public static final String LABELVALUE = "label_value";

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            /** set offset to latest */
            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "expAttribution")
                            .uid("kfk-expAttribution-source");

            SingleOutputStreamOperator<String> out =
                    input.flatMap(new clickAttributionFormat())
                            .keyBy(
                                    new KeySelector<AttributionInfo, String>() {
                                        @Override
                                        public String getKey(AttributionInfo value)
                                                throws Exception {
                                            return value.getDevice_id();
                                        }
                                    })
                            .flatMap(new clickAttributionProcess());

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            //            sinkProp.setProperty("max.request.size", String.valueOf(KFKMSGMAXSIZE));
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> kafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(AdmExposureAttributionKfkLabel.class.toString());
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class clickAttributionFormat
            extends RichFlatMapFunction<String, AttributionInfo> {

        @Override
        public void flatMap(String value, Collector<AttributionInfo> out) throws Exception {
            try {
                JSONObject inputJson = JSONObject.parseObject(value);
                if (inputJson != null) {
                    String hfqIdfa = inputJson.getString(HFQIDFA);
                    String hfqOaid = inputJson.getString(HFQOAID);

                    hfqIdfa = formatIdfaOaid(hfqIdfa);
                    hfqOaid = formatIdfaOaid(hfqOaid);

                    String deviceId = (hfqIdfa == null ? hfqOaid : hfqIdfa);
                    String unique_channel = inputJson.getString(CHANNEL);

                    if (deviceId != null && unique_channel != null) {
                        long recv_time = inputJson.getLongValue(RECVTIME);
                        if (recv_time >= STARTIME) {
                            String hfq_advstId = inputJson.getString(HFQADVSTID);
                            String hfq_cmpgId = inputJson.getString(HFQCMPGID);
                            String hfq_aid = inputJson.getString(HFQAID);
                            String hfq_clkId = inputJson.getString(HFQCLKID);
                            out.collect(
                                    new AttributionInfo(
                                            unique_channel,
                                            recv_time,
                                            deviceId,
                                            hfq_advstId,
                                            hfq_cmpgId,
                                            hfq_aid,
                                            hfq_clkId));
                        }
                    }
                }
            } catch (Exception e) {
                LOG.error("clickAttributionRichFlatMap: " + e.toString());
                LOG.error("clickAttributionRichFlatMap input: " + value);
            }
        }
    }

    private static class clickAttributionProcess
            extends RichFlatMapFunction<AttributionInfo, String> {
        MapStateDescriptor<Long, JSONObject> mStateDesc;
        MapState<Long, JSONObject> clkAttributMState;
        SimpleDateFormat sdf;

        @Override
        public void open(Configuration parameters) throws Exception {
            StateTtlConfig ttlConfig =
                    StateTtlConfig.newBuilder(Time.hours(24)) // dt 按天分区 只存24小时
                            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                            .cleanupInRocksdbCompactFilter(100000L)
                            .build();

            mStateDesc =
                    new MapStateDescriptor<>("clickAttributionDesc", Long.class, JSONObject.class);
            mStateDesc.enableTimeToLive(ttlConfig);

            clkAttributMState = getRuntimeContext().getMapState(mStateDesc);

            sdf = new SimpleDateFormat(DATEFORMAT);
        }

        @Override
        public void flatMap(AttributionInfo input, Collector<String> out) throws Exception {
            try {
                LOG.debug("##input: " + input.toString());
                String unique_channel = input.getUnique_channel();
                long receive_time = input.getReceive_time();
                // dt_time 秒级时间戳
                long dt_time = getStartTimeOfDay(receive_time, sdf);

                // 分区判断
                LOG.debug(
                        "contains dt_time: " + dt_time + " " + clkAttributMState.contains(dt_time));
                if (clkAttributMState.contains(dt_time)) {
                    JSONObject stateDtJson = clkAttributMState.get(dt_time);
                    if (stateDtJson != null) {
                        if (!stateDtJson.containsKey(unique_channel + 0)) {
                            // new channel
                            input.setRecv_cnt(1);
                            // rt_type0 最早信息
                            stateDtJson.put(unique_channel + 0, input);
                            // rt_type1 最晚信息
                            stateDtJson.put(unique_channel + 1, input);
                            // 更新MapState
                            clkAttributMState.put(dt_time, stateDtJson);

                            // 发送下游 需要debug确认
                            input.setRt_type(0);
                            LOG.debug("new channel debug0: " + input.toString());
                            directOutputJson(input, dt_time, out);

                            input.setRt_type(1);
                            LOG.debug("new channel debug1: " + input.toString());
                            directOutputJson(input, dt_time, out);
                        } else {
                            AttributionInfo earlyInfo =
                                    (AttributionInfo) stateDtJson.get(unique_channel + 0);
                            AttributionInfo stateEarlyBak = new AttributionInfo(earlyInfo);
                            // 有流入cnt就+1
                            int recvCnt = earlyInfo.getRecv_cnt() + 1;
                            long earlyTime = earlyInfo.getReceive_time();

                            if (receive_time < earlyTime) {
                                input.setRecv_cnt(recvCnt);
                                stateDtJson.put(unique_channel + 0, input);
                            } else {
                                earlyInfo.setRecv_cnt(recvCnt);
                                stateDtJson.put(unique_channel + 0, earlyInfo);
                            }
                            AttributionInfo outAttribution =
                                    (AttributionInfo) stateDtJson.get(unique_channel + 0);
                            outAttribution.setRt_type(0);

                            // 发送下游 需要debug确认
                            LOG.debug("old debug0: " + outAttribution.toString());
                            cmpOutputJson(outAttribution, stateEarlyBak, dt_time, out);

                            AttributionInfo lateInfo =
                                    (AttributionInfo) stateDtJson.get(unique_channel + 1);
                            AttributionInfo stateLateBak = new AttributionInfo(lateInfo);
                            long lateTime = lateInfo.getReceive_time();

                            if (receive_time > lateTime) {
                                input.setRecv_cnt(recvCnt);
                                stateDtJson.put(unique_channel + 1, input);
                            } else {
                                lateInfo.setRecv_cnt(recvCnt);
                                stateDtJson.put(unique_channel + 1, lateInfo);
                            }
                            AttributionInfo outLateAttribution =
                                    (AttributionInfo) stateDtJson.get(unique_channel + 1);
                            outLateAttribution.setRt_type(1);
                            LOG.debug("old debug1: " + outLateAttribution.toString());
                            cmpOutputJson(outLateAttribution, stateLateBak, dt_time, out);

                            clkAttributMState.put(dt_time, stateDtJson);
                        }
                    }
                } else {
                    // new dt
                    JSONObject dtTimeJson = new JSONObject();

                    input.setRecv_cnt(1);
                    // rt_type0 最早信息
                    dtTimeJson.put(unique_channel + 0, input);
                    // rt_type1 最晚信息
                    dtTimeJson.put(unique_channel + 1, input);
                    // 更新MapState
                    clkAttributMState.put(dt_time, dtTimeJson);

                    // 发送下游 需要debug确认
                    input.setRt_type(0);
                    LOG.debug("new dt debug0: " + input.toString());
                    directOutputJson(input, dt_time, out);

                    input.setRt_type(1);
                    LOG.debug("new dt debug1: " + input.toString());
                    directOutputJson(input, dt_time, out);
                }

            } catch (Exception e) {
                LOG.error("excep: " + e.toString());
                LOG.error("excep input: " + input);
            }
        }

        private void cmpOutputJson(
                AttributionInfo value, AttributionInfo state, long dt_time, Collector<String> out) {
            LOG.debug("@@value:" + value.toString());
            LOG.debug("@@state:" + state.toString());
            String device_id = value.getDevice_id();
            String unique_channel = value.getUnique_channel();

            JSONObject outJson = new JSONObject();
            // unique key
            outJson.put(DEVICEID, device_id);
            outJson.put(DTTIME, dt_time);
            outJson.put(CHANNEL, unique_channel);
            outJson.put(RTTYPE, value.getRt_type());

            if (value.getRecv_cnt() != state.getRecv_cnt()) {
                outJson.put(LABELNAME, RECVCNT);
                outJson.put(LABELVALUE, value.getRecv_cnt());
                LOG.debug("#cmpOut: " + outJson.toString());
                out.collect(outJson.toString());
            }

            String input_adv_id = value.getHfq_advertiser_id();
            String st_adv_id = state.getHfq_advertiser_id();
            LOG.debug("input_adv_id: " + (input_adv_id == null ? "null" : input_adv_id));
            LOG.debug("st_adv_id: " + (st_adv_id == null ? "null" : st_adv_id));
            if (input_adv_id != null && st_adv_id != null) {
                if (!input_adv_id.equals(st_adv_id)) {
                    outJson.put(LABELNAME, HFQADVSTID);
                    outJson.put(LABELVALUE, input_adv_id);
                    LOG.debug("#cmpOut1: " + outJson.toString());
                    out.collect(outJson.toString());
                }
            } else if (input_adv_id != st_adv_id) {
                outJson.put(LABELNAME, HFQADVSTID);
                outJson.put(LABELVALUE, input_adv_id);
                LOG.debug("#cmpOut2: " + outJson.toString());
                out.collect(outJson.toString());
            }

            String input_cmpg_id = value.getHfq_campaign_id();
            String st_cmpg_id = state.getHfq_campaign_id();
            LOG.debug("input_cmpg_id: " + (input_cmpg_id == null ? "null" : input_cmpg_id));
            LOG.debug("st_cmpg_id: " + (st_cmpg_id == null ? "null" : st_cmpg_id));
            if (input_cmpg_id != null && st_cmpg_id != null) {
                if (!input_cmpg_id.equals(st_cmpg_id)) {
                    outJson.put(LABELNAME, HFQCMPGID);
                    outJson.put(LABELVALUE, input_cmpg_id);
                    LOG.debug("#cmpOut1: " + outJson.toString());
                    out.collect(outJson.toString());
                }
            } else if (input_cmpg_id != st_cmpg_id) {
                outJson.put(LABELNAME, HFQCMPGID);
                outJson.put(LABELVALUE, input_cmpg_id);
                LOG.debug("#cmpOut2: " + outJson.toString());
                out.collect(outJson.toString());
            }

            String input_aid = value.getHfq_aid();
            String st_aid = state.getHfq_aid();
            LOG.debug("input_aid: " + (input_aid == null ? "null" : input_aid));
            LOG.debug("st_aid: " + (st_aid == null ? "null" : st_aid));
            if (input_aid != null && st_aid != null) {
                if (!input_aid.equals(st_aid)) {
                    outJson.put(LABELNAME, HFQAID);
                    outJson.put(LABELVALUE, input_aid);
                    LOG.debug("#cmpOut1: " + outJson.toString());
                    out.collect(outJson.toString());
                }
            } else if (input_aid != st_aid) {
                outJson.put(LABELNAME, HFQAID);
                outJson.put(LABELVALUE, input_aid);
                LOG.debug("#cmpOut2: " + outJson.toString());
                out.collect(outJson.toString());
            }

            String input_clk_id = value.getHfq_click_id();
            String st_clk_id = state.getHfq_click_id();
            LOG.debug("input_clk_id: " + (input_clk_id == null ? "null" : input_clk_id));
            LOG.debug("st_clk_id: " + (st_clk_id == null ? "null" : st_clk_id));
            if (input_clk_id != null && st_clk_id != null) {
                if (!input_clk_id.equals(st_clk_id)) {
                    outJson.put(LABELNAME, HFQCLKID);
                    outJson.put(LABELVALUE, value.getHfq_click_id());
                    LOG.debug("#cmpOut1: " + outJson.toString());
                    out.collect(outJson.toString());
                }
            } else if (value.getHfq_click_id() != st_clk_id) {
                outJson.put(LABELNAME, HFQCLKID);
                outJson.put(LABELVALUE, value.getHfq_click_id());
                LOG.debug("#cmpOut2: " + outJson.toString());
                out.collect(outJson.toString());
            }
        }

        private void directOutputJson(AttributionInfo value, long dt_time, Collector<String> out) {
            String device_id = value.getDevice_id();
            String unique_channel = value.getUnique_channel();

            JSONObject outJson = new JSONObject();
            // unique key
            outJson.put(DEVICEID, device_id);
            outJson.put(DTTIME, dt_time);
            outJson.put(CHANNEL, unique_channel);
            outJson.put(RTTYPE, value.getRt_type());

            outJson.put(LABELNAME, RECVCNT);
            outJson.put(LABELVALUE, value.getRecv_cnt());
            //            LOG.debug("#direct: " + outJson.toString());
            out.collect(outJson.toString());

            outJson.put(LABELNAME, HFQADVSTID);
            outJson.put(LABELVALUE, value.getHfq_advertiser_id());
            //            LOG.debug("#direct: " + outJson.toString());
            out.collect(outJson.toString());

            outJson.put(LABELNAME, HFQCMPGID);
            outJson.put(LABELVALUE, value.getHfq_campaign_id());
            //            LOG.debug("#direct: " + outJson.toString());
            out.collect(outJson.toString());

            outJson.put(LABELNAME, HFQAID);
            outJson.put(LABELVALUE, value.getHfq_aid());
            //            LOG.debug("#direct: " + outJson.toString());
            out.collect(outJson.toString());

            outJson.put(LABELNAME, HFQCLKID);
            outJson.put(LABELVALUE, value.getHfq_click_id());
            //            LOG.debug("#direct: " + outJson.toString());
            out.collect(outJson.toString());
        }
    }

    private static long getStartTimeOfDay(long now, SimpleDateFormat sdf) {

        String date = sdf.format(new Date(now));
        long res = 0;
        try {
            res = sdf.parse(date).getTime() / 1000;
        } catch (Exception e) {
            LOG.error("sdf parse date error! time: " + now);
            e.printStackTrace();
        }
        return res;
    }

    //    private static String formatIdfaOaid(String input) {
    //        String res = null;
    //        if(input != null && !"00000000-0000-0000-0000-000000000000".equals(input)
    //                && !"00000000000000000000000000000000".equals(input)
    //                && !"".equals(input)
    //                && !"__IDFA__".equals(input) && !"__OAID__".equals(input)) {
    //            res = input;
    //        }
    //        return res;
    //    }

    private static String formatIdfaOaid(String input) {
        String res = null;
        if (input != null
                && !"9F89C84A559F573636A47FF8DAED0D33"
                        .equals(input) // '00000000-0000-0000-0000-000000000000'
                && !"CD9E459EA708A948D5C2F5A6CA8838CF"
                        .equals(input) // '00000000000000000000000000000000'
                && !"D41D8CD98F00B204E9800998ECF8427E".equals(input) // ''
                && !"6E0D8DA38B72B0A55A28F0764389A1E7".equals(input) // '__IDFA__'
                && !"A261C265DF3A6D24B5AD3CC1DDAFD8E6".equals(input) // '__OAID__'
                && !"A5452BD63942BFCA9CD8E1C09223383E".equals(input) // '__IDFA2__'
                && !"6C3E226B4D4795D518AB341B0824EC29".equals(input) // 'NULL'
                && input.length() == 32) {
            res = input;
        }
        return res;
    }

    //    private static JSONObject initPropJson(String propFile) throws IOException {
    //        InputStream resourceAsStream =
    // AdmClickAttributionKfkLabel.class.getClassLoader().getResourceAsStream(propFile);
    //        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
    //        JSONObject res = new JSONObject();
    //        Iterator<Object> keysIterator = reader.getProperties().keySet().iterator();
    //        while (keysIterator.hasNext()) {
    //            String key = (String) keysIterator.next();
    //            if(key.equals(CAPTIME)) {
    //                res.put(key, -1);
    //            } else {
    //                res.put(key, "initV");
    //            }
    //        }
    //        return res;
    //    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(DEVICEID);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.debug(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }

    public static void test(String[] args) {
        String input = "9F89C84A559F573636A47FF8DAED0E33";
        System.out.println("Tst input res: " + formatIdfaOaid(input));
    }
}
